// Copyright 2013 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.google.api.ads.dfa.matchtables;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;

/**
 * Reads in Data Transfer files handling international characters and issues with new lines.
 *
 * @author Moshe Kahan
 */
public class DataTransferFileScanner {

  private static final int BUF_SIZE = 8192;

  private DataTransferFileParser parser;
  private String regexDelimiter;
  private StringBuilder stringBuilder;
  private byte[] buf;
  private int count;
  private int pos;

  DataTransferFileScanner(DataTransferFileParser parser, String regexDelimiter) {
    this.parser = parser;
    this.regexDelimiter = regexDelimiter;
    stringBuilder = new StringBuilder();

    // Using byte buffer since BufferedInputStream is too slow.
    buf = new byte[BUF_SIZE];
    count = -1;
    pos = 0;
  }

  // Read all DT Files line by line. Each file is sent to parseDTFile.
  void parseDTFiles(String[] dtFileNames) throws DataTransferFileException {
    try {
      for (String dtFileName : dtFileNames) {
        FileInputStream fileInputStream = new FileInputStream(dtFileName);
        try {
          parseDTFileStream(fileInputStream, dtFileName);
        } finally {
          fileInputStream.close();
        }
      }
    } catch (FileNotFoundException e) {
      throw new DataTransferFileException("Data Transfer file not found", e);
    } catch (IOException e) {
      throw new DataTransferFileException("IOException when closing Data Transfer file", e);
    }
  }

  void parseDTFileStream(InputStream inputStream, String dtFileName)
      throws DataTransferFileException {
    System.out.println("Parsing DT File: " + dtFileName);
    BufferedInputStream bufferedInputStream = new BufferedInputStream(inputStream);
    try {
      parseStream(bufferedInputStream);
    } finally {
      try {
        bufferedInputStream.close();
      } catch (IOException e) {
        throw new DataTransferFileException("IOException when closing Data Transfer stream", e);
      }
    }
  }

  private void parseStream(InputStream inputStream) throws DataTransferFileException {
    parser.clearHeaders();
    final int million = 1000000;
    try {
      String line = headerLine(inputStream);
      if (line != null && !line.isEmpty()) {
        // Assuming first line is header
        int delimitersNeeded = parser.parseHeader(line) - 1;
        long x = 0;
        char delimiter = regexDelimiter.charAt(0);
        do {
          try {
            ++x;
            // read in the line
            line = nextLine(inputStream, delimiter, delimitersNeeded);
            if (line != null && !line.isEmpty()) {
              // process the line
              parser.parse(line);
            }
          } catch (DataTransferFileException e) {
            // Log bad line and skip
            System.out.println("Possibly corrupt DT file on line #" + x + " with message: "
                + e.getMessage());
          } catch (NumberFormatException e) {
            // Log bad line and skip
            System.out.println("Corrupt DT file line #" + x + " with contents: " + line);
          }
          if (x % million == 0) {
            // Progress in form of [XXX million] records printed every 10mil records
            System.out.print("\r[" + x / million +  " million lines parsed]");
          }
        } while (line != null);
        if (x >= million) {
          System.out.println();
        }
      }
    } catch (IOException e) {
      throw new DataTransferFileException("IOException when reading Data Transfer stream", e);
    }
  }

  private String headerLine(InputStream inputStream) throws IOException {
    stringBuilder.setLength(0);
    int read = -1;
    while ((read = inputStream.read()) != -1 && read != '\n') {
      stringBuilder.append(((char) read));
    }
    return stringBuilder.toString();
  }

  private String nextLine(InputStream inputStream, char delimiter, int delimitersNeeded)
      throws DataTransferFileException, IOException {
    stringBuilder.setLength(0);
    int read = -1;
    int delimiterCount = 0;
    boolean middleColumns = false;

    // \n stops the read unless we have seen some columnDelimiters but not all
    // of the column delimiters we expected. In that case, keep reading even
    // if we read in a line delimiter.
    do {
      if (pos >= count) {
        pos = 0;
        if ((count = inputStream.read(buf, 0, BUF_SIZE)) < 0) {
          break;
        }
      }
      read = buf[pos++] & 0xff;
      // We only care about Time and numeric ID fields, so this cast from byte
      // to char is fine no matter the encoding of the character data.
      if (read == delimiter) {
        ++delimiterCount;
        if (delimiterCount < delimitersNeeded) {
          middleColumns = true;
        } else {
          middleColumns = false;
        }
      }
      if (!middleColumns && read == '\n') {
        break;
      } else {
        stringBuilder.append(((char) read));
      }
    } while (true);

    String line = stringBuilder.toString();
    if (read == -1 && line.isEmpty()) {
      return null;
    }
    if (delimiterCount == 0) {
      // Assume this was an invalid line delimiter in the first column.
      // Log and continue
      String message = "No delimiters found";
      if (line != null && !line.isEmpty()) {
        message = message + " on line: " + line;
      }
      throw new DataTransferFileException(message);
    }
    return line;
  }
}
